package com.jast.flume;

import cn.hutool.json.JSONUtil;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * @author Jast
 * @description 自定义拦截器
 * @date 2022-03-17 09:20
 */
public class ETLInterceptor implements Interceptor {

    private final String separator = "\t";
    private final String defaultValue = "";
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        //拦截kafka时，这里获取到的Head，是Kafka元信息,包括key，topic，partition，offset等信息，e.g. {topic=ZW_WB_1010000009, partition=0, offset=5425705, key={"dataType":"ZW_WB_","fileName":"yc_1644913124_11086","fullLinkId":"3e9fc439e42c4059b46e5d5664d08bf8","lineNum":17,"subFullLinkId":"77dc3a9680ee4d1caeb6719d87e8d3e5","totalCount":50}, timestamp=1648642188763}
        Map<String, String> headers = event.getHeaders();
        String topic = headers.getOrDefault("topic",defaultValue);
        String partition = headers.getOrDefault("partition",defaultValue);
        String offset = headers.getOrDefault("offset",defaultValue);
        String key = headers.getOrDefault("key",defaultValue);
        long currentTimeMillis = System.currentTimeMillis();
        //System.out.println("拦截器head:"+headers);
        byte[] body = event.getBody();
        String text = new String(body, StandardCharsets.UTF_8);
        if(JSONUtil.isJson(text)){
            text = topic+separator+text+separator+partition+separator+offset+separator+key+separator+currentTimeMillis;
            System.out.println("text:"+text);
            event.setBody(text.getBytes());
            return event;
        }
        //System.out.println("非json格式过滤掉");
        return null;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        Iterator<Event> iterator = list.iterator();
        while (iterator.hasNext()){
            Event next = iterator.next();
            if(intercept(next)==null){
                iterator.remove();
            }
        }
        return list;
    }

    @Override
    public void close() {

    }


    public static class Builder implements Interceptor.Builder{

        @Override
        public Interceptor build() {
            return new ETLInterceptor();
        }
        @Override
        public void configure(Context context) {

        }
    }
}
